package com.syf.kit.soa.core.send.protocol.client;

import com.syf.kit.soa.core.SyfRpcMsg;
import com.syf.kit.soa.core.send.protocol.TcpClient;
import com.syf.kit.soa.core.send.protocol.TcpResponseCoordinator;
import com.syf.kit.soa.core.send.protocol.TcpResponseLatch;
import com.syf.kit.soa.core.send.protocol.exception.RpcResponseTimeoutException;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

@Slf4j
public class LongTcpConnection implements TcpClient {
    private Bootstrap bootstrap;

    /**
     * 连接建立的超时时间，单位：毫秒
     */
    private Long connTimeout;

    private AttributeKey<Boolean> USABLE_ATTR_KEY;

    private ChannelFuture cf;

    public LongTcpConnection(Bootstrap bootstrap, Long connTimeout, AttributeKey<Boolean> USABLE_ATTR_KEY) {
        this.bootstrap = bootstrap;
        this.connTimeout = connTimeout;
        this.USABLE_ATTR_KEY = USABLE_ATTR_KEY;
    }

    /**
     * 建立连接
     */
    public void connect(String host, int port) {
        try {
            cf = bootstrap.connect(host, port);

            if (!cf.await(connTimeout, TimeUnit.MILLISECONDS)) {
                throw new RuntimeException("建立连接超时，目标：[" + host + ":" + port + "]");
            }

            if (cf.isSuccess() && cf.channel().isActive()) {
                cf.channel().attr(USABLE_ATTR_KEY).set(true);
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 是否可用
     */
    public boolean isUsable() {
        return cf.channel().attr(USABLE_ATTR_KEY).get();
    }

    /**
     * 关闭
     */
    public void close() {
        cf.channel().close();
    }

    @Override
    public SyfRpcMsg send(SyfRpcMsg reqMsg, long reqTimeout) {
        if (!cf.channel().isActive()) {
            throw new RuntimeException("TCP CONNECTION IS NOT ACTIVE");
        }

        cf.channel().writeAndFlush(reqMsg);
        TcpResponseLatch<SyfRpcMsg> sendLatch = TcpResponseCoordinator.getInstance().loadSendLatch(reqMsg.getReqUUID());
        try {
            return sendLatch.getResponse(reqTimeout);
        } catch (RpcResponseTimeoutException e) {
            throw new RuntimeException(e);
        }
    }
}
